/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.ybk.e2p.yb.service;

import cn.ac.iie.di.ban.ycl.ybk.e2p.yb.tools.fastutil.FastPo;
import java.io.File;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;

/**
 *
 * @author zhangyun
 */
public class FastUploadService {

    private static Logger logger = Logger.getLogger(FastUploadService.class);
    long complete = 0;//完成上传的文件数
    public String status = "";
    public String desc = "";

    public HashMap<String, FastPo> upload(String localPath, String timeout, List<Map<String, String>> dataFileList) {
        HashMap<String, FastPo> retStringMap = new HashMap<String, FastPo>();
        List<String> finalFileList = new ArrayList<String>();
        for (int i = 0; i < dataFileList.size(); i++) {
            Map<String, String> str = dataFileList.get(i);
            String lastFilename = str.get("name");
            String realName = lastFilename.substring(lastFilename.lastIndexOf("/") + 1, lastFilename.length());
            String dataFileName = localPath + lastFilename;
            String okFileName = localPath + lastFilename + ".ok";
            File dataf = new File(dataFileName);
            File okf = new File(okFileName);
//            if (okf.exists() && dataf.exists()) {
//                if (!realName.endsWith(".json") && !realName.endsWith(".ok")) {
            if (dataf.exists()) {
                if (!realName.endsWith(".json")) {
                    finalFileList.add(lastFilename);
                }
            }
        }

//        File fileRoot = new File(localPath);
//        if (!fileRoot.isDirectory()) {
//            logger.error(" localPath is not dir");
//            return null;
//        }
//        String[] allFileList = fileRoot.list();
//        //match file
//        List<String> fileList = new ArrayList<String>();
//        List<String> okFileList = new ArrayList<String>();
//        List<String> finalFileList = new ArrayList<String>();
//        for (int i = 0; i < allFileList.length; i++) {
//            if (allFileList[i].endsWith(".ok")) {
//                okFileList.add(allFileList[i]);
//            } else {
//                fileList.add(allFileList[i]);
//            }
//        }
//        logger.info(" file total number: " + fileList.size());
//        for (int i = 0; i < fileList.size(); i++) {
//            for (int j = 0; j < okFileList.size(); j++) {
//                if ((fileList.get(i) + ".ok").equals(okFileList.get(j))) {
//                    if (!fileList.get(i).endsWith(".json")) {
//                        Map<String, String> str = dataFileList.get(i);
//                        String lastFilename = str.get("name");
//                        String realName = lastFilename.substring(lastFilename.lastIndexOf("/") + 1, lastFilename.length());
//                        if (realName.equals(fileList.get(i))) {
//                            finalFileList.add(fileList.get(i));
//                        }
//                    }
//                }
//            }
//        }
        Integer totalSize = finalFileList.size();
        logger.info(" file upload total number: " + finalFileList.size());
        for (int i = 0; i < finalFileList.size(); i++) {
            logger.info(" file[" + i + "]: " + finalFileList.get(i));
        }

        List<String> finalList = new ArrayList<String>();
        if (finalFileList.size() == 0) {//没有要采集的文件
            logger.error(" no files for collecting");
        }
        for (int i = 0; i < finalFileList.size(); i++) {
            finalList.add(finalFileList.get(i));
            if (finalList.size() >= 1000) {
                logger.error(" >=1000");
                uploadPoolTask(finalList, localPath, timeout, retStringMap);
                finalList.clear();
            }
            if (i == finalFileList.size() - 1 && !finalList.isEmpty()) {
                uploadPoolTask(finalList, localPath, timeout, retStringMap);
            }
        }
        Long completeInt = complete;

        //分析状态信息
        if (status.contains("fail") && status.contains("sucess")) {
            status = "fail";
        }
        if (status.contains("fail") && !status.contains("sucess")) {
            status = "fail";
        }
        if (!status.contains("fail") && status.contains("sucess")) {
            status = "sucess";
        }

        return retStringMap;

    }

    public HashMap<String, FastPo> uploadPoolTask(List<String> finalFileList, String localPath, String timeout, HashMap<String, FastPo> retStringMap) {

        //线程池
        long begin = System.currentTimeMillis();
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
                TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(finalFileList.size()),
                new ThreadPoolExecutor.DiscardOldestPolicy());
        for (int i = 0; i < finalFileList.size(); i++) {
//            threadPool.execute(new FastThreadPoolTask(localPath + "/" + finalFileList.get(i), subjobid, pathList, this));
            new FastThreadPoolTask(localPath + finalFileList.get(i), retStringMap, this).run();
            logger.info(" return FastDFS path number: " + retStringMap.size());
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException ex) {
            logger.error(" sleep error " + ex.getMessage(), ex);
        }
        while (true) {//直至所有线程结束，才继续
            int activeThread = threadPool.getActiveCount();
            if (activeThread == 0) {
                break;
            } else {
                try {
                    long end = System.currentTimeMillis();
                    long middle = (end - begin) / 1000;
                    long tmpTimeout = Long.parseLong(timeout);
                    if (middle < tmpTimeout) {
                        logger.info(" runtime:" + middle + " timeout:" + timeout);
                        Thread.sleep(3000);
                    } else {
                        logger.error(" upload is killed, because of timeout!");
                        changeStatus("fail");
                        changeDesc("timeout");
                        threadPool.shutdown();
                        return null;
                    }
                    Thread.sleep(1000);
                } catch (InterruptedException ex) {
                    logger.error(" sleep error " + ex.getMessage(), ex);
                }
            }
        }
        threadPool.shutdown();
//        if(finalFileList.size()> retStringMap.size() ){
//            retStringMap = null;
//        }
        return retStringMap;
    }

//统计上传成功文件个数
    public synchronized void plusSucessUploadFileNumber() {
        complete = complete + 1;
    }

    //修改上传状态
    public synchronized void changeStatus(String st) {
        status = status + "|" + st;
    }

    //修改上传描述
    public synchronized void changeDesc(String pDesc) {
        desc = desc + "|" + pDesc;
    }
}
